package com.celeritymq.consumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;

import com.celeritymq.listeners.CelerityMessageListener;
import com.celeritymq.message.CelerityMessage;

public abstract class CelerityMessageConsumer implements MessageConsumer {
	private CelerityMessageListener messageListener;
	private String messageSelector;
	private BlockingQueue<CelerityMessage> messageQueue = new LinkedBlockingQueue<CelerityMessage>();

	protected CelerityMessageConsumer(String messageSelector) {
		this.messageSelector = messageSelector;
	}

	@Override
	public void close() throws JMSException {
		messageQueue.clear();
		messageQueue = null;
		messageListener = null;
	}

	@Override
	public MessageListener getMessageListener() throws JMSException {
		return messageListener;
	}

	@Override
	public String getMessageSelector() throws JMSException {
		return messageSelector;
	}

	@Override
	public Message receive() throws JMSException {
		try {
			return messageQueue.take();
		} catch (InterruptedException e) {

		}
		return null;
	}

	@Override
	public Message receive(long timeout) throws JMSException {
		try {
			return messageQueue.poll(timeout, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public Message receiveNoWait() throws JMSException {
		return messageQueue.poll();
	}

	@Override
	public void setMessageListener(MessageListener messageListener) throws JMSException {
		if (messageListener instanceof CelerityMessageListener) {
			this.messageListener = (CelerityMessageListener) messageListener;
		} else {
			throw new IllegalArgumentException(messageListener + " is not a CelerityMessageListener!");
		}
	}

}
